package com.shxzz.easy12306.services.ticket.canal;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import com.alibaba.fastjson.JSON;
import com.shxzz.easy12306.framework.starter.cache.StringRedisTemplateProxy;
import com.shxzz.easy12306.framework.starter.cache.toolkit.CacheUtil;
import com.shxzz.easy12306.framework.starter.common.mq.event.CanalBinlogEvent;
import com.shxzz.easy12306.framework.starter.designpattern.strategy.AbstractExecuteStrategy;
import com.shxzz.easy12306.services.ticket.common.constant.RedisKeyConstant;
import com.shxzz.easy12306.services.ticket.common.enums.CanalExecuteStrategyMark;
import com.shxzz.easy12306.services.ticket.common.enums.SeatStatusEnum;
import com.shxzz.easy12306.services.ticket.config.TicketRemainCacheUpdateProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * @author shxzz
 * @create 2024-03-18-23:45
 */
@Slf4j
@RequiredArgsConstructor
@Component
@ConditionalOnProperty(
        prefix = TicketRemainCacheUpdateProperties.PREFIX,
        name = TicketRemainCacheUpdateProperties.TICKET_UPDATE_TYPE_NAME,
        havingValue = TicketRemainCacheUpdateProperties.BINLOG_UPDATE_TYPE
)
public class TicketRemainCacheUpdateHandler implements AbstractExecuteStrategy<CanalBinlogEvent, Void> {
    /**
     * 座位状态字段
     */
    public static final String SEAT_STATUS_FIELD = "seat_status";
    public static final String TRAIN_ID_FIELD = "train_id";
    public static final String START_STATION_FIELD = "start_station";
    public static final String END_STATION_FIELD = "end_station";
    public static final String SEAT_TYPE_FIELD = "seat_type";

    private final StringRedisTemplateProxy stringRedisTemplateProxy;

    @Override
    public String mark() {
        return CanalExecuteStrategyMark.T_SEAT.getActualTable();
    }

    @Override
    public void execute(CanalBinlogEvent message) {
        // 过滤出座位状态变更为 AVAILABLE 或 LOCKED 的信息
        List<Map<String, Object>> currentDataList = new ArrayList<>();
        List<Map<String, Object>> oldDataList = new ArrayList<>();
        for (int i = 0; i < message.getOld().size(); i++) {
            Map<String, Object> oldDataMap = message.getOld().get(i);
            if (oldDataMap.containsKey(SEAT_STATUS_FIELD)) {
                Map<String, Object> currentDataMap = message.getData().get(i);
                if (CharSequenceUtil.equalsAny(
                        currentDataMap.get(SEAT_STATUS_FIELD).toString(),
                        String.valueOf(SeatStatusEnum.AVAILABLE.getCode()),
                        String.valueOf(SeatStatusEnum.LOCKED.getCode())
                )) {
                    oldDataList.add(oldDataMap);
                    currentDataList.add(currentDataMap);
                }
            }
        }
        if (CollUtil.isEmpty(currentDataList)) {
            return;
        }

        Map<String, Map<Integer, Integer>> cacheChangeKeyMap = new HashMap<>();
        for (int i = 0; i < currentDataList.size(); i++) {
            Map<String, Object> currentDataMap = currentDataList.get(i);
            Map<String, Object> oldDataMap = oldDataList.get(i);
            String seatStatus = oldDataMap.get(SEAT_STATUS_FIELD).toString();
            int increment = Objects.equals(seatStatus, SeatStatusEnum.AVAILABLE.getCode().toString()) ? -1 : 1;
            String trainId = currentDataMap.get(TRAIN_ID_FIELD).toString();
            String startStation = currentDataMap.get(START_STATION_FIELD).toString();
            String endStation = currentDataMap.get(END_STATION_FIELD).toString();
            String cacheKey = CacheUtil.buildKey(RedisKeyConstant.TRAIN_REMAIN_TICKET_PREFIX, trainId, startStation, endStation);
            Map<Integer, Integer> seatTypeMap = cacheChangeKeyMap.get(cacheKey);
            if (CollUtil.isEmpty(seatTypeMap)) {
                seatTypeMap = new HashMap<>();
            }
            Integer seatType = Integer.parseInt(currentDataMap.get(SEAT_TYPE_FIELD).toString());
            Integer delta = seatTypeMap.getOrDefault(seatType, 0);
            seatTypeMap.put(seatType, delta + increment);
            cacheChangeKeyMap.put(cacheKey, seatTypeMap);
        }
        cacheChangeKeyMap.forEach((cacheKey, seatTypeMap) ->
                log.info("binlog修改余量缓存，余量key:{},变化量：{}", cacheKey, JSON.toJSONString(seatTypeMap))
        );
        StringRedisTemplate stringRedisTemplate = stringRedisTemplateProxy.getInstance();
        stringRedisTemplate.executePipelined((RedisCallback<String>) connection -> {
            cacheChangeKeyMap.forEach((cacheKey, seatTypeMap) -> seatTypeMap.forEach((seatType, delta) -> connection.hashCommands()
                    .hIncrBy(cacheKey.getBytes(),
                            String.valueOf(seatType).getBytes(),
                            delta))
            );
            return null;
        });
    }
}
